/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2023 All Rights Reserved.
 */
package com.example.spider.cluster.msg;

import com.example.spider.task.AbstractSpiderTask;
import jakarta.annotation.Nonnull;
import org.jetbrains.annotations.NotNull;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.support.MessageBuilder;

import java.time.Duration;

/**
 * https://juejin.cn/post/6956104044326682655
 * 消息的分割（1->n）：Splitor
 * 消息的聚合（1->n）：Aggregator
 * 消息的过滤（m->n）: Filter
 * 消息的过滤（m->n）: Dispatcher
 * <p>
 * 生产者只创建 excahnges、
 * 消费者创建 queue/binder（根据配置option）只有消费者才会创建队列，queue名称为：<bindingName><destination>.<group>
 * <p>
 * https://www.jianshu.com/p/ef19d04d1558
 * https://docs.spring.io/spring-cloud-stream/docs/3.2.4/reference/html/spring-cloud-stream-binder-rabbit.html#_rabbitmq_consumer_properties
 * https://cloud.spring.io/spring-cloud-stream-binder-rabbit/spring-cloud-stream-binder-rabbit.html#_rabbit_producer_properties
 *
 * @author lym
 */
public class MsgSender {

    private final StreamBridge streamBridge;

    /**
     * 爬虫待执行任务队列名称（可延迟）
     */
    private final String spiderTaskQueueName;

    /**
     * 爬虫结果队列名称
     */
    private final String spiderResultQueueName;

    /**
     * 处理器结果队列名称
     */
    private final String handlerResultQueueName;

    public MsgSender(StreamBridge streamBridge, String spiderTaskQueueName, String spiderResultQueueName, String handlerResultQueueName) {
        this.streamBridge = streamBridge;
        this.spiderTaskQueueName = spiderTaskQueueName;
        this.spiderResultQueueName = spiderResultQueueName;
        this.handlerResultQueueName = handlerResultQueueName;
    }


    public boolean addSpiderTask(@Nonnull AbstractSpiderTask<?> spiderTask, Duration delayDuration) {
        return sendMsg(spiderTaskQueueName,
                MessageBuilder.withPayload(UnionSpiderMsg.builder().task(spiderTask).build())
                        //最多支持延迟 Duration.ofDays(49) 超过则立即消费
                        .setHeader("x-delay", delayDuration == null ? "0" : delayDuration.toMillis())
                        .setHeader("task_type", spiderTask.getClass().getSimpleName())
        );
    }

    /**
     * 可以定义标准result：包含 task、http response body、response header 信息
     */
    public boolean addSpiderResult(@Nonnull AbstractSpiderTask<?> spiderTask, @Nonnull String spiderResult) {
        return sendMsg(spiderResultQueueName,
                MessageBuilder.withPayload(UnionSpiderMsg.builder().task(spiderTask).spiderResult(spiderResult).build())
                        .setHeader("task_type", spiderTask.getClass().getSimpleName())
        );
    }

    /**
     * 和需求强关联，建议case by case
     * 而非定义标准result：包含 task、http 信息、body 信息
     */
    public boolean addHandlerResult(@Nonnull Object handlerResult) {
        return sendMsg(handlerResultQueueName, MessageBuilder.withPayload(handlerResult));
    }

    private boolean sendMsg(String queueName, @NotNull MessageBuilder<?> msgBuilder) {
        return streamBridge.send(queueName, msgBuilder.build());
    }

}